package com.bff.gaia.runtime.maqy;

import com.esotericsoftware.minlog.Log;
import com.bff.gaia.api.common.functions.RichMapPartitionFunction;
import com.bff.gaia.api.common.typeutils.TypeComparator;
import com.bff.gaia.api.common.typeutils.TypeComparatorFactory;
import com.bff.gaia.util.Collector;

import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.List;

/**
 * Build RangeBoundaries with input records. First, sort the input records, and then select
 * the boundaries with percent interval.
 *
 * @param <T>
 */
public class PercentRangeBoundaryBuilder<T> extends RichMapPartitionFunction<T, Object[][]> {

	private int parallelism;

	private final TypeComparatorFactory<T> comparatorFactory;

	private ArrayList<Integer> percentPerChannel = new ArrayList<>();//这里不能static

	public ArrayList<Integer> getPercentPerChannel() {
		return this.percentPerChannel;
	}

	//注意这里如果不是一个元素一个元素添加，size会对应不上
	public void setPercentPerChannel(ArrayList<Integer> percentPerChannel) {
		this.getPercentPerChannel().clear();
		for(int i : percentPerChannel){
			this.getPercentPerChannel().add(i);
		}
		//PercentRangeBoundaryBuilder.percentPerChannel = new ArrayList(Arrays.asList(percentPerChannel));
	}

	public PercentRangeBoundaryBuilder(TypeComparatorFactory<T> comparator, int parallelism) {
		this.comparatorFactory = comparator;
		this.parallelism = parallelism;
	}

	@Override
	public void mapPartition(Iterable<T> values, Collector<Object[][]> out) throws Exception {
		List<Object> broadcastVariable = getRuntimeContext().getBroadcastVariable("Proportions");
		if (broadcastVariable == null || broadcastVariable.size() != 1) {
			throw new RuntimeException("PercentRangeBoundaryBuilder require a single Proportions as broadcast input.");
		}
		List<Integer> proportions = (List<Integer>) broadcastVariable.get(0);

		final TypeComparator<T> comparator = this.comparatorFactory.createComparator();
		List<T> sampledData = new ArrayList<>();
		for (T value : values) {
			sampledData.add(value);
		}
		Collections.sort(sampledData, new Comparator<T>() { //这里进行了排序
			@Override
			public int compare(T first, T second) {
				return comparator.compare(first, second);
			}
		});

		int boundarySize = parallelism - 1;
		Object[][] boundaries = new Object[boundarySize][];//第二维是因为key可能是多维的
		String logInfo = "";
		if (sampledData.size() > 0) {
			int numKey = comparator.getFlatComparators().length;
			if (proportions != null && proportions.size() == parallelism) {
				logInfo = String.format("Percent Split: (boundaries size=%d)\n", parallelism - 1);

				// 按比例分配
				int splitNum = 0;
				for (int i = 1; i < parallelism; i++) {//每个并行度（分区）一个边界值
					double avgRange = sampledData.size() * (double)proportions.get(i-1) / 100;
					splitNum  += (int)avgRange;

					// 防止下标越界（最后一个比例分量为0会出现这种情况）
					if (splitNum >= sampledData.size()) {
						splitNum = sampledData.size() - 1;
					}

					T record = sampledData.get(splitNum);//计算得到靠近段尾的采样记录作为边界界定标准
					Object[] keys = new Object[numKey];
					comparator.extractKeys(record, keys, 0);
					boundaries[i-1] = keys; //考虑一下，如果一个key相同的数据很多，两次的boundaries都相同怎么办呢？估计要看后续的map？

					logInfo += String.format("%5s#%d: %s\n", "", i, keys.toString());
				}
			} else {
				logInfo = String.format("Average Split: (boundaries size=%d)\n", parallelism - 1);

				// 平均分配
				double avgRange = sampledData.size() / (double) parallelism;//计算拆分的段,即每个段有多少元素

				for (int i = 1; i < parallelism; i++) {//每个并行度（分区）一个边界值
					T record = sampledData.get((int) (i * avgRange));//计算得到靠近段尾的采样记录作为边界界定标准
					Object[] keys = new Object[numKey];
					comparator.extractKeys(record, keys, 0);
					boundaries[i-1] = keys; //考虑一下，如果一个key相同的数据很多，两次的boundaries都相同怎么办呢？估计要看后续的map？

					logInfo += String.format("%5s#%d: %s\n", "", i, keys.toString());
				}
			}
		}

		Log.info("hfy", "[PercentRangeBoundaryBuilder] " + logInfo.substring(0, logInfo.length() - 1));

		out.collect(boundaries);
	}
}

